package cn.seecoder.ai.serviceImpl;

import cn.seecoder.ai.dao.DeploymentRepository;
import cn.seecoder.ai.enums.DeploymentStateEnum;
import cn.seecoder.ai.enums.TrainingStateEnum;
import cn.seecoder.ai.exception.AIExternalException;
import cn.seecoder.ai.exception.AIInternalException;
import cn.seecoder.ai.model.po.Deployment;
import cn.seecoder.ai.model.vo.DeploymentVO;
import cn.seecoder.ai.model.vo.FileInfoVO;
import cn.seecoder.ai.model.vo.ModelVO;
import cn.seecoder.ai.model.vo.PredictionVO;
import cn.seecoder.ai.service.*;
import cn.seecoder.ai.utils.POJOConverter;
import lombok.extern.slf4j.Slf4j;
import org.apache.spark.ml.PipelineModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * @author fanyanpeng
 * @date 2023/4/22 20:26
 */
@Slf4j
@Service
public class DeploymentServiceImpl implements DeploymentService {

    // 上传文件
    @Autowired
    FileInfoService fileInfoService;

    // 触发文件分析
    @Autowired
    FileFieldService fileFieldService;

    @Autowired
    DeploymentRepository deploymentRepository;

    @Autowired
    PredictionScalaService predictionScalaService;

    @Autowired //需要判断模型的状态，以决定是否能够部署该模型
    ModelService modelService;

    @Autowired //在已加载模型的情况下调用 “预测模型”
    PredictionService predictionService;

    @Autowired
    ApplicationContext applicationContext;

    @Value("${cn.seecoder.ai.ml.deployment-prefix}")
    private String deploymentPrefix;

    private Map<Integer, PipelineModel> deployedModelMap = new ConcurrentHashMap<>();





    private Deployment toPO(DeploymentVO deploymentVO){
        return (Deployment) POJOConverter.convert(deploymentVO);
    }

    private DeploymentVO toVO(Deployment deployment){
        return (DeploymentVO) POJOConverter.convert(deployment);
    }




    @Scheduled(initialDelay = 2000,fixedDelay = 600000)
    //定时每十秒钟确认当前内存和数据库部署状态一致
    @Override
    public void checkDeploymentStateSync(){
        log.info("触发部署状态检查");

        log.info("判断是否有已经部署的任务，但是目前内存中没有记录");
        List<Deployment> deploymentList = deploymentRepository.findByDeploymentState(DeploymentStateEnum.SUCCESS);
        for(Deployment deployment:deploymentList){
            if(!deployedModelMap.containsKey(deployment.getId())){
                log.info("内存中没有记录，但是数据库中有记录，将其加入内存");
                asyncDeploy(deployment.getId());
                log.info("部署成功，{}",deployment.getId());
            }
        }
        log.info("current online deployment{}",deployedModelMap.keySet().stream().map(this::getDeployment).collect(Collectors.toList()).toString());

    }



    /**
     * 获取用户的部署历史
     *
     * @param userId
     * @return java.util.List<cn.seecoder.ai.model.vo.DeploymentVO>
     * @author fanyanpeng
     * @date 2023/4/22 20:17
     */
    @Override
    public List<DeploymentVO> getDeploymentVOList(Integer userId) {
        return deploymentRepository.findByUserId(userId).stream().map(this::toVO).collect(Collectors.toList());
    }

    /**
     * 获取模型的部署历史
     *
     * @param userId
     * @param modelId
     * @return java.util.List<cn.seecoder.ai.model.vo.DeploymentVO>
     * @author fanyanpeng
     * @date 2023/4/22 20:17
     */
    @Override
    public List<DeploymentVO> getDeploymentVOList(Integer userId, Integer modelId) {
        return deploymentRepository.findByUserIdAndModelId(userId,modelId).stream().map(this::toVO).collect(Collectors.toList());
    }

    /**
     * 获取单个部署的详细信息
     *
     * @param userId
     * @param predictionId
     * @return cn.seecoder.ai.model.vo.DeploymentVO
     * @author fanyanpeng
     * @date 2023/4/22 20:19
     */
    @Override
    public DeploymentVO getDeploymentVO(Integer userId, Integer predictionId) {
        return deploymentRepository.findById(predictionId).map(this::toVO).get();
    }


    /**
     * @param deploymentVO
     * @return
     */
    @Override
    public DeploymentVO renewDeploymentVO(DeploymentVO deploymentVO) {
        Deployment deployment = toPO(deploymentVO);
        renew(deployment);
        return toVO(deployment);
    }

    @Async
    public void asyncDeploy(Integer deploymentId){
        Deployment deployment = getDeployment(deploymentId);

        try{

            //设置正在部署
            deployment.setDeploymentState(DeploymentStateEnum.ON_GOING);
            deploymentRepository.save(deployment);

            Integer userId = deployment.getUserId();
            Integer modelId = deployment.getModelId();
            ModelVO modelVO = modelService.getModelVO(userId,modelId);

            // 若模型的状态不是训练成功，就被不能部署
            if(!modelVO.getTrainingState().equals(TrainingStateEnum.SUCCESS)){
                throw AIInternalException.modelNotDeployable(modelId,modelVO.getTrainingState());
            }

            // 认为模型部署成功了，进行部署工作
            String modelUri = modelVO.getModelUri();
            PipelineModel pipelineModel = PipelineModel.load(modelUri);

            // 将模型放入本地映射表
            this.deployedModelMap.put(deploymentId,pipelineModel);
            deployment.setDeploymentState(DeploymentStateEnum.SUCCESS);
            deployment.setFinishTime(new Date());

            //部署的url前缀
            deployment.setDeploymentUrl(deploymentPrefix+deploymentId+"/use-with-file");
            deploymentRepository.save(deployment);

        }catch (Exception e){   // 遇到错误，更新信息并保存
            deployment.setDeploymentState(DeploymentStateEnum.FAIL);
            deploymentRepository.save(deployment);
            throw e;
        }
    }




    /**
     * 取消部署一个模型
     *
     * @param userId
     * @param deploymentId
     * @return cn.seecoder.ai.model.vo.DeploymentVO
     * @author fanyanpeng
     * @date 2023/4/22 20:19
     */
    @Override
    public DeploymentVO unDeploy(Integer userId, Integer deploymentId) {

        Deployment deployment = getDeployment(deploymentId);

        // 移除流水线模型
        this.deployedModelMap.remove(deploymentId);

        //设置模型未部署
        deployment.setDeploymentState(DeploymentStateEnum.WAITING);
        deployment.setDeploymentUrl(null);
        deployment.setFinishTime(null);

        deploymentRepository.save(deployment);

        return getDeploymentVO(userId, deploymentId);
    }

    /**
     * 重新部署一个模型：重新获取部署对应的模型，进行加载
     *
     * @param userId
     * @param deploymentId
     * @return cn.seecoder.ai.model.vo.DeploymentVO
     * @author fanyanpeng
     * @date 2023/4/22 20:19
     */
    @Override
    public DeploymentVO reDeploy(Integer userId, Integer deploymentId) {
        Deployment deployment=getDeployment(deploymentId);
        renew(deployment);
        asyncDeploy(deploymentId);
        return getDeploymentVO(userId, deploymentId);
    }

    /**
     * 使用部署的模型：取出模型，调用预测服务，返回
     *
     * @param userId
     * @param deploymentId
     * @param fileInfoId
     * @return cn.seecoder.ai.model.vo.DeploymentVO
     * @author fanyanpeng
     * @date 2023/4/22 20:19
     */
    @Override
    public PredictionVO useDeployedModel(Integer userId, Integer deploymentId, Integer fileInfoId) {
        if(!this.deployedModelMap.containsKey(deploymentId)){
            throw AIExternalException.noSuchDeployment(deploymentId);
        }
        Deployment deployment = getDeployment(deploymentId);
        Integer modelId =deployment.getModelId();
        PipelineModel pipelineModel = this.deployedModelMap.get(deploymentId);
        return predictionService.predictWithPipeLineModelSync(userId,modelId,pipelineModel,fileInfoId,null);
    }


    /**
     * 使用部署的模型：取出模型，调用预测服务，返回
     *
     * @param deploymentId 部署id
     * @param file         待预测文件
     * @return cn.seecoder.ai.model.vo.DeploymentVO
     * @author fanyanpeng
     * @date 2023/4/22 20:19
     */
    @Override
    public PredictionVO useDeployedModelWithFile(Integer deploymentId, MultipartFile file) {
        FileInfoVO fileInfoVO = fileInfoService.addFile(getFileName(),0,file);
        Integer fileInfoId = fileInfoVO.getId();
        fileFieldService.analysisField(0,fileInfoId);
        if(!this.deployedModelMap.containsKey(deploymentId)){
            throw AIExternalException.noSuchDeployment(deploymentId);
        }
        Deployment deployment = getDeployment(deploymentId);
        Integer modelId =deployment.getModelId();
        PipelineModel pipelineModel = this.deployedModelMap.get(deploymentId);
        return predictionService.predictWithPipeLineModelSync(0,modelId,pipelineModel,fileInfoId,null);
    }

    private String getFileName(){
        return "deployment-use-"+new Date().getTime()+".csv";
    }



    /**
     * 初始化部署情况并保存到数据库
     * @author   fanyanpeng
     * @date 2023/4/23 5:10
     * @param deployment
     * @return void
     */
    private void renew(Deployment deployment){
        // 设置状态为等待部署
        deployment.setDeploymentState(DeploymentStateEnum.WAITING);
        deployment.setCreatedTime(new Date());
        deployment.setFinishTime(null);
        deployment.setDeploymentUrl(null);

        deploymentRepository.save(deployment);

    }

    private Deployment getDeployment(Integer deploymentId){
        return deploymentRepository.findById(deploymentId).get();
    }



}
